 #include "../Headers/dis_img_pro.h"
void count_files(const char *dir_path, st *pSt)
{
    DIR *dir;
    struct dirent *entry;
    int count = 0;

    dir = opendir(dir_path);
    if (dir == NULL)
    {
        perror("opendir");
        return;
    }
    while ((entry = readdir(dir)) != NULL)
    {
        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
        {
            continue;
        }

        ImgEnqueue(pSt, entry->d_name, count);
        count++;
    }
    closedir(dir);
}

int ImgEnqueue(st *pImgQueue, char *filename, int ID)
{
    int len = strlen(filename) + 1;
    up_img *pUp_img = (up_img *)malloc(sizeof(up_img));
    if (NULL == pUp_img)
    {
        perror("malloc");
        return -1;
    }

    strcpy(pUp_img->buf, filename);
    pUp_img->ID = ID;
    printf("%d\n", pImgQueue->size);
    if (pImgQueue->size == 0)
    {

        pImgQueue->pFront = pUp_img;
        pImgQueue->pRear = pUp_img;
    }
    else
    {

        pImgQueue->pRear->pNext = pUp_img;
        pImgQueue->pRear = pUp_img;
    }
    ++pImgQueue->size;

    return 0;
}

int ImgDequeue(st *pImgQueue)
{
    up_img *pCur = pImgQueue->pFront;
    pImgQueue->pFront = pCur->pNext;
    free(pCur);
    --pImgQueue->size;
    return 0;
}
st *pImgQueueInit()
{
    st *pImgQueue = (st *)malloc(sizeof(st));
    pImgQueue->pFront = NULL;
    pImgQueue->pRear = NULL;
    pImgQueue->size = 0;
    return pImgQueue;
}
int transFile(int netFd,char*filename){
    char path[1024] = {0};
    sprintf(path, "runs/detect/predict/%s", filename);
    int fd = open(path, O_RDWR);
    ERROR_CHECK(fd,-1,"open");
    puts(path);
    //int ret = send(netFd,"file1",5,0);
    //ERROR_CHECK(ret,-1,"send");
    train_t train;
    train.length = strlen(filename);
    strcpy(train.buf,filename);
    printf("filename:%s\n",filename);
    int ret = send(netFd,&train,sizeof(train.length)+train.length,MSG_NOSIGNAL);
    struct stat statbuf;

    ret = fstat(fd,&statbuf);
    ERROR_CHECK(ret,-1,filename);
    train.length = 4;//车厢是4个字节 int
    int fileSize = statbuf.st_size; // 长度转换成int
    memcpy(train.buf,&fileSize,sizeof(int));//int存入小火车
    send(netFd,&train,train.length+sizeof(train.length),MSG_NOSIGNAL);

    
    /*
    char *p = (char *)mmap(NULL,fileSize,PROT_READ|PROT_WRITE,MAP_SHARED,fd,0);
    ERROR_CHECK(p,MAP_FAILED,"mmap");
    send(netFd,p,fileSize,MSG_NOSIGNAL);
    */
    sendfile(netFd,fd,NULL,fileSize);
    train.length = 0;
    ret = send(netFd,&train,sizeof(train.length)+train.length,MSG_NOSIGNAL);
    return 0;
}
void uploadingTask(int net_fd)
{

    puts("up");
    st *imgs = pImgQueueInit();
    size_t i = 0;
    char buf[1024];
    sprintf(buf, "/home/leao/work/sdSys/project/distributed-image-processing/dis_img_pro/runs/detect/predict/");
    count_files(buf, imgs);
    puts("coutfile ok");
    printf("imgs=%d\n", imgs->size);
    send(net_fd, TASK_DOWN, 1024, 0);
    recv(net_fd, buf, 1024, 0);
    if (0 == strcmp(buf, TASK_ON))
    {
        puts(buf);

        // 在线校验
        int cout = imgs->size;
        for (int i = 0; i < cout; i++)
        {
            bzero(buf, 1024);
            puts(imgs->pFront->buf);
            send(net_fd, TASK_ON, 1024, 0);
            transFile(net_fd, imgs->pFront->buf);
            printf("i=%d\n", i);
            puts(imgs->pFront->buf);
            printf("now_imgs=%d\n", imgs->size);
            ImgDequeue(imgs);
            bzero(buf, 1024);
            printf("now_imgs=%d\n", imgs->size);
            while (1)
            {
                recv(net_fd, buf, 1024, 0);
                if (0 == strcmp(buf, TASK_ON)) // 多次任务递交校验
                {
                    break;
                }
            }
        }
        send(net_fd, TASK_SND_OVER, 1024, 0);
    }
    else
    {
        puts("服务器转发异常");
    }
    puts("任务分发完成开始执行");
}
int recvn(int sockFd, void *pstart, int len)
{
    int total = 0;
    int ret;
    char *p = (char *)pstart;
    while (total < len)
    {
        ret = recv(sockFd, p + total, len - total, 0);
        total += ret;
    }
    return 0;
}
int transmitTask(int sockFd,int destFd)
{
    // 先获取文件名
    char name[1024] = {0};
    int dataLength;
    // int ret = recv(sockFd,&dataLength,sizeof(int),MSG_WAITALL);
    int ret = recvn(sockFd, &dataLength, sizeof(int));
    ERROR_CHECK(ret, -1, "recv");
    printf("%d\n",dataLength);
    ret = send(destFd,&dataLength,4,MSG_NOSIGNAL);
    ERROR_CHECK(ret,-1,"send_dest");


    // ret = recv(sockFd,name,dataLength,MSG_WAITALL);
    ret = recvn(sockFd, name, dataLength);
    ERROR_CHECK(ret, -1, "recv");
    send(destFd,name,1024,0);
 ERROR_CHECK(ret,-1,"send_dest");

 
    //  char new_name[2048] = {0};
    // sprintf(new_name, "/online_task/%s", name);
    // puts(new_name);
    // int fd = open(new_name, O_RDWR | O_CREAT | O_TRUNC, 0666);
    // ERROR_CHECK(fd, -1, "open");
    int fileSize;
    recvn(sockFd, &dataLength, sizeof(int));
    ret = send(destFd,&dataLength,4,MSG_NOSIGNAL);
    ERROR_CHECK(ret,-1,"send_dest");

    recvn(sockFd, &fileSize, dataLength);
    ret = send(destFd,&fileSize,4,MSG_NOSIGNAL);
    ERROR_CHECK(ret,-1,"send_dest");
    printf("fileSize = %d\n", fileSize);

    //char buf[1000] = {0};
    time_t timeBeg, timeEnd;
    timeBeg = time(NULL);
    // int pipefds[2];
    // pipe(pipefds);
    int total = 0;
    
    // while (total < fileSize)
    // {
    //     int ret = splice(sockFd, NULL, pipefds[1], NULL, 4096, SPLICE_F_MORE);
    //     total += ret;
    //     usleep(2000000);
    //     splice(pipefds[0], NULL, fd, NULL, ret, SPLICE_F_MORE);
    // }

    
    // char *p = (char *)mmap(NULL,fileSize,PROT_READ|PROT_WRITE,MAP_SHARED,fd,0);
    // ERROR_CHECK(p,MAP_FAILED,"mmap");
    char *p = calloc(fileSize,1);
    recvn(sockFd,p,fileSize);
    ret = send(destFd,p,fileSize,0);
    ERROR_CHECK(ret,-1,"send_dest");

    recvn(sockFd, &dataLength, sizeof(int));
    ret = send(destFd,p,fileSize,0);
    ERROR_CHECK(ret,-1,"send_dest");

    printf("dataLength = %d\n", dataLength);
    timeEnd = time(NULL);
    printf("total time = %ld\n", timeEnd - timeBeg);

    free(p);
    return 0;
}
